
package com.atguigu.day7

import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object OrderTimeoutDetect {
  case class OrderEvent(ordirId :String,eventType:String,eventTime:Long)

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val stream = env.fromElements(
      OrderEvent("order_1", "create", 2000L),
      OrderEvent("order_2", "create", 3000L),
      OrderEvent("order_2", "pay", 4000L)
    )
      .assignAscendingTimestamps(_.eventTime)
      .keyBy(_.ordirId)


      val pattern = Pattern
        .begin[OrderEvent]("create")
        .where(_.eventType.equals("create"))
        .next("pay")
        .where(_.eventType.equals("pay"))
        .within(Time.seconds(5))

      val patternStream = CEP.pattern(stream,pattern)

//
//    patternStream
//      .select((pattern:scala.collection.Map[String,Iterable[OrderEvent]])=> {
//        val event = pattern("create").iterator.next()
//        "数据有"+event.ordirId
//
//      }).print()

//      //用来输出超时订单信息
      val OrderTimeoutOutputTag = new OutputTag[String]("timeout")

      //这个匿名函数用来处理超时的检测
      val timeoutFunc = (map:scala.collection.Map[String,Iterable[OrderEvent]],ts:Long,out:Collector[String])=>{
        print("在"+ts +"ms 之前没有支付")
        val orderCreat = map("create").iterator.next()
        out.collect("超时订单的ID为"+orderCreat.ordirId)

      }
    //这个匿名函数用来处理支付成功的检测
      val selectFunc: (collection.Map[String, Iterable[OrderEvent]], Collector[String]) => Unit = (map:scala.collection.Map[String,Iterable[OrderEvent]], out:Collector[String])=>{
        val orderPay = map("pay").iterator.next()
        out.collect("订单ID为"+orderPay.ordirId+"支付成功")
      }


    val detectStream: DataStream[String] = patternStream
      //flatSelect方法接收柯里化参数
      //第一个参数：检测出的超时信息发送到的侧输出标签
      //第二个参数：用来处理超时信息的函数
      //第三个参数：用来处理create和pay匹配成功的信息
      .flatSelect(OrderTimeoutOutputTag)(timeoutFunc)(selectFunc)
    //打印成功信息
    detectStream.print()
    //打印超时信息
    detectStream.getSideOutput(OrderTimeoutOutputTag).print()

    env.execute()


  }

}
